📐 API, Specs & Structured Outputs Cheat Sheet

A quick-reference guide for HTTP status codes, asynchronous connection pooling using httpx, tenacity retry handlers, OpenAPI 3.0 YAML templates, Pydantic v2 schemas, and Instructor structured extraction loops.

---

🌐 HTTP Status Codes Reference

Code	Status	Meaning	Agent Action Plan
200	OK	Request completed successfully.	Parse the body payload.
201	Created	Resource created successfully (e.g. database write).	Extract unique resource IDs.
400	Bad Request	Malformed request body or parameter mismatch.	Log exception; do not retry immediately.
401	Unauthorized	Missing or invalid API key / Bearer Token.	Suspend execution and report credentials error.
403	Forbidden	Valid credentials, but lacks permissions for path.	Suspend execution; log access bounds block.
404	Not Found	Endpoint path or primary resource does not exist.	Verify path structures or query values.
429	Too Many Requests	Server Rate Limit exceeded (Throttling active).	Backoff and retry with randomized jitter.
500	Internal Error	Server crashed internally.	Retry with exponential backoff (max 3 runs).
503	Service Unavail	Server offline or overloaded.	Retry with exponential backoff.

---

⚡ Asynchronous httpx Pool with tenacity Retries

Write non-blocking, rate-limit resilient API clients:

import asyncio
import httpx
from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type

### 1. Configure the retry policy for HTTP exceptions and 429 status codes
@retry(
    wait=wait_random_exponential(min=1, max=10), # Exponential base + randomized jitter
    stop=stop_after_attempt(3), # Terminate execution after 3 attempts
    retry=retry_if_exception_type(httpx.HTTPStatusError),
    reraise=True
)
async def fetch_llm_payload(client: httpx.AsyncClient, prompt: str) -> dict:
    url = "https://api.provider.com/v1/generate"
    headers = {"Authorization": "Bearer SECURE_TOKEN"}
    payload = {"model": "gemini-1.5-flash", "prompt": prompt}
    
    response = await client.post(url, json=payload, headers=headers)
    
    # Automatically raise HTTPStatusError if status is 4xx or 5xx (e.g. 429)
    response.raise_for_status()
    return response.json()

async def main():
    # 2. Configure a pooled connection client
    limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
    async with httpx.AsyncClient(limits=limits) as client:
        try:
            data = await fetch_llm_payload(client, "Translate Hello World to French")
            print("Successfully extracted payload:", data)
        except Exception as e:
            print("API loop failed after maximum retries:", e)

if __name__ == "__main__":
    asyncio.run(main())

---

📐 OpenAPI 3.0 YAML Template (Spec-Driven)

openapi: 3.0.0
info:
  title: Agent CRM API
  description: API for creating and querying corporate contacts in a CRM tool database.
  version: 1.0.0
servers:
  - url: http://127.0.0.1:8000/v1
paths:
  /contacts:
    post:
      summary: Add a new contact
      operationId: addContact
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/Contact'
      responses:
        '201':
          description: Contact created successfully.
components:
  schemas:
    Contact:
      type: object
      required: [name, email]
      properties:
        name:
          type: string
          description: Full legal name of contact.
        email:
          type: string
          format: email
          description: Valid primary business email address.

---

📐 Pydantic v2 Schemas & JSON Schema Compiler

import json
from pydantic import BaseModel, Field, field_validator

### 1. Define model class with validation parameters
class Contact(BaseModel):
    name: str = Field(..., description="Full legal name of contact.")
    email: str = Field(..., description="Valid business email.")
    priority: int = Field(default=1, ge=1, le=5, description="Priority rating (1 to 5)")

    # Custom validation rule
    @field_validator("email")
    @classmethod
    def validate_email_domain(cls, value: str) -> str:
        if "@" not in value:
            raise ValueError("Invalid email format.")
        return value

### 2. Compile dynamic JSON Schema conforming to draft-07 standards
schema = Contact.model_json_schema()
print(json.dumps(schema, indent=2))

---

🛡️ Instructor Structured outputs & Self-Correction Retries

Enforce type-safe JSON extractions out of messy unstructured logs:

import instructor
import google.generativeai as genai
from pydantic import BaseModel, Field

### 1. Define target extraction class
class ActionItem(BaseModel):
    task: str = Field(..., description="Actionable checklist item.")
    owner: str = Field(..., description="Person assigned to task.")
    due_date: str = Field(..., description="Due date mentioned in transcript.")

class IncidentReport(BaseModel):
    summary: str = Field(..., description="Short summary of incident.")
    sentiment_score: int = Field(..., ge=1, le=10, description="1=Angry, 10=Happy")
    checklist: list[ActionItem] = Field(..., description="Action items identified.")

### 2. Patch the Gemini client to support type-safe extraction models
client = instructor.from_gemini(
    client=genai.Client(),
    mode=instructor.Mode.GEMINI_JSON # Enforces Structured outputs
)

transcript = "John said we must deploy the database patch by Friday. I am very frustrated with the delays! Dave will coordinate test suites."

### 3. Dispatched typed extraction loop with self-correcting retry rules
report = client.chat.completions.create(
    model="gemini-1.5-flash",
    response_model=IncidentReport, # Enforces strict validation matching class
    messages=[{"role": "user", "content": transcript}],
    max_retries=2 # If Pydantic fails, automatically feeds exception trace back to LLM to retry
)

print(report.model_dump_json(indent=2))